package com.dtwave.cfstreaming.filter.function;

import com.alibaba.fastjson.JSONObject;
import com.dtwave.cfstreaming.connection.ElasticConnector;
import com.dtwave.cfstreaming.connection.HbaseConnector;
import com.dtwave.utils.JsonUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class NotNullFilterFunction extends RichFilterFunction<JSONObject> {
    private final static Logger LOGGER = LoggerFactory.getLogger(NotNullFilterFunction.class);

    RestHighLevelClient esClient;
    Connection hbaseClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        //获取flink主程序中的参数内容并打开需要的连接数据库的客户端
        ParameterTool Parameters = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        hbaseClient = HbaseConnector.getConnection(Parameters.get("HbaseQuorum", "aaa"),
                Parameters.get("hbase_host_key", "hbase_host"),
                Parameters.getInt("连接池个数", 30));
        //esClient同上
        esClient = ElasticConnector.getClient("es连接地址");
    }


    /**
     * true会传递到下游，false不会传递到下游
     *
     * @param jsonObject
     * @return
     * @throws Exception
     */
    @Override
    public boolean filter(JSONObject jsonObject) throws Exception {
        String onType = JsonUtils.getString(jsonObject, true, "操作类型（I&U）");

        if (!"I".equals(onType) && !"U".equals(onType)){
            return false;
        }
        LOGGER.debug("正常数据传输！");
        //过滤满足条件的数据
        return true;
    }

    @Override
    public void close() throws Exception {
        if (hbaseClient != null) {
            hbaseClient.close();
        }
        if (esClient != null) {
            esClient.close();
        }
    }
}
